package com.xdxc.service.consumer;

import cn.hutool.core.lang.Assert;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xdxc.exception.CustomException;
import com.xdxc.model.dto.UserAnswerDto;
import com.xdxc.utils.MergeWindowManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.Base64;
import java.util.Map;

@Component
@RocketMQMessageListener(
        topic = "Exam",
        consumerGroup = "GRADING_ANSWER_SAVE_GROUP",
        consumeMode = ConsumeMode.ORDERLY // 关键配置
)
@Slf4j
public class AnswerSaveConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private MergeWindowManager mergeWindowManager;
    @Autowired
    private static final ObjectMapper mapper = new ObjectMapper()
            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

    @Override
    public void onMessage(MessageExt message) {
        final String msgId = message.getMsgId();
        final String keys = message.getKeys();
        try {
            log.info("▶▶ 开始处理消息 msgId={}, keys={}", msgId, keys);
            // 1. 消息解析（带异常捕获）
            UserAnswerDto dto = parseMessage(message);

            // 2. 合并处理（带耗时监控）
            long start = System.currentTimeMillis();
            mergeWindowManager.accumulate(dto);

            log.debug("◀◀ 合并处理完成 keys={}, 耗时={}ms", keys, System.currentTimeMillis()-start);
        } catch (Exception e) {
            // 可重试异常（如网络抖动）
            log.error("⚠ 消费失败 [msgId={}, keys={}], 剩余重试次数={}. 错误: {}",
                    msgId, keys, message.getReconsumeTimes(), e.getMessage());
            throw new RuntimeException(e);
        }

    }
    @PreDestroy
    public void onShutdown() {
        log.info("消费者关闭，执行合并窗口持久化...");
        mergeWindowManager.shutdown();
    }

    public UserAnswerDto parseMessage(MessageExt message) {
        try {
            // 1. 基础元数据提取
            Map<String, String> properties = message.getProperties();
            String messageKey = properties.get(RocketMQHeaders.KEYS);
            // 2. 反序列化消息体
            UserAnswerDto dto = mapper.readValue(message.getBody(), UserAnswerDto.class);
            // 3. 补充系统字段
//            dto.setExtFields(new HashMap<>());
//            dto.getExtFields().put("mqMsgId", message.getMsgId());
//            dto.getExtFields().put("mqBornTimestamp", message.getBornTimestamp());
            // 4. 数据校验
            validateAnswerDTO(dto);
            return dto;
        } catch (Exception e) {
            log.error("消息解析失败 msgId={}, body={}", message.getMsgId(), message.getBody(), e);
            String rawBody = Base64.getEncoder().encodeToString(message.getBody());
            log.error("消息解析失败 [msgId={}], 原始数据(base64): {}",
                    message.getMsgId(), rawBody);
            throw new CustomException(400,"消息格式错误");
        }
    }


    private void validateAnswerDTO(UserAnswerDto dto) {
//        Assert.notNull(dto.get(), "考试ID不能为空");
        Assert.notNull(dto.getCheckId(), "用户ID不能为空");
        Assert.notNull(dto.getQuId(), "题目ID不能为空");

        // 根据题型验证answerData格式
//        switch (QuestionType.of(dto.getQuestionType())) {
//            case SINGLE_CHOICE:
//                JSONObject.parseObject(dto.getAnswerData()).getString("option");
//                break;
//            case MULTI_CHOICE:
//                JSONArray options = JSONObject.parseObject(dto.getAnswerData()).getJSONArray("options");
//                Assert.isTrue(options.size() <= 5, "多选题选项过多");
//                break;
//            // 其他题型校验...
//        }
    }
}